package org.example.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * @author Tang
 * @classname PushConsumer
 * @description [  ]
 * @date 2022/11/24 00:19
 */
@Slf4j
public class PushOrderlyConsumer {

    public static void main(String[] args) throws Exception{

        DefaultMQPushConsumer consumer1 = new DefaultMQPushConsumer("default");
        consumer1.setNamesrvAddr("localhost:9876");
        consumer1.setConsumeTimeout(30000);

        new Thread(new Task("test-topic",consumer1)).start();

        Thread.sleep(1000);
        consumer1.start();
        System.in.read();
    }



    public static class Task implements Runnable{

        private final String name;

        private final DefaultMQPushConsumer consumer;

        public Task(String name,DefaultMQPushConsumer consumer) {
            this.name = name;
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                consumer.subscribe(this.name,"*");
                consumer.setSuspendCurrentQueueTimeMillis(5000);
                consumer.setMaxReconsumeTimes(5);
                consumer.registerMessageListener((MessageListenerOrderly) (msgs, consumeConcurrentlyContext) -> {
                    for (MessageExt msg : msgs) {
                        String msgBody = new String(msg.getBody());
                        System.out.println(Task.this.name + " Topic: " + msg.getTopic() +
                                ",Tags: " + msg.getTags() +
                                ",QueueId: " + msg.getQueueId() +
                                ",msg: " + msgBody + " " +
                                "header: " + msg.getProperties()
                        );
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                });

                log.info("consumer.start...");
                System.in.read();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


}
